package biz.ws.receive.kafka;

import biz.ws.WsConfig;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.client.naming.NacosNamingService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.Closeable;
import java.io.IOException;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.locks.LockSupport;

/**
 * @创建人 wsdc
 * @时间 2021/4/2
 * @描述 <li>    HA high available
 * <li>    定时检测另外另外一个节点，如果宕机了，那么去监听宕机节点的kafka 分区
 * <li>    基于 Nacos 服务发现来做判断
 */
@Component
@Slf4j
public class HAScheduler extends Thread implements InitializingBean, Closeable {
    @Resource
    NacosNamingService nacosNamingService;

    @Resource
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    @Resource
    WsConfig wsConfig;

    boolean shutdown = false;
    boolean sharedLinkIsWorking = false;

    int count = 0;

    MessageListenerContainer sharedLinkNode;

    @Override
    public void afterPropertiesSet() throws Exception {
        //MessageListenerContainer c1 = kafkaListenerEndpointRegistry.getListenerContainer("kafkaMessageReceiveMain");
        sharedLinkNode = kafkaListenerEndpointRegistry.getListenerContainer("kafkaMessageReceiveTemp");

        //  自身属于一个线程
        if (wsConfig.isSharedTemp()) {
            start();
        }
    }


    @Override
    public void run() {
        while (!shutdown) {
            LockSupport.parkNanos(wsConfig.getCheckLinkInLineTime() * 1000l * 1000 * 1000);

            try {
                //  使用nacos来判断  后续可以调整成 ping/pong 接口来上线
                List<Instance> list = nacosNamingService.getAllInstances("ws1");
                Optional<Instance> sharedLink1 = list.stream()
                        .filter(k -> {
                            String shared = k.getMetadata().get("shared");
                            return Objects.equals(shared, wsConfig.getSharedLink() + "");
                        })
                        .findFirst();

                if (sharedLinkNode == null) {
                    continue;
                }

                //  输出一条日志
                if (count++ % 100 == 0) {
                    log.info("HAScheduler run 扫描关联节点的在线状态");
                }

                if (sharedLink1.isPresent() && sharedLinkIsWorking) {
                    //  如果节点存在 但是此时正在工作 那么停止
                    sharedLinkNode.pause();
                    sharedLinkIsWorking = false;
                    log.info("HAScheduler run 主节点已经上线，从节点退出...");
                    continue;
                }

                if (!sharedLink1.isPresent() && !sharedLinkIsWorking) {
                    if (!sharedLinkNode.isRunning()) {
                        sharedLinkNode.start();
                        sharedLinkIsWorking = true;
                        log.info("HAScheduler run 主节点已经下线，丛节点开启 工作");
                        continue;
                    }

                    sharedLinkIsWorking = true;
                    sharedLinkNode.resume();
                    continue;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void close() throws IOException {
        shutdown = true;
    }
}
